package streamAPI.source;

import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.concurrent.TimeUnit;

/**
 * 自定义支持多并行度 的 Source
 *
 * @author yue.cao
 * @since 09-28-2020
 */
public class MyParallelSource implements ParallelSourceFunction<Long> {

	private Long count = 1L;
	private boolean isRunning = true;

	@Override
	public void run(SourceContext<Long> ctx) throws Exception {
		while (isRunning) {
			ctx.collect(count++);
			TimeUnit.SECONDS.sleep(1);
		}
	}
	@Override
	public void cancel() {
		isRunning = false;
	}
}
